package com.janlei.service;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import software.amazon.awssdk.core.async.AsyncResponseTransformer;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.services.s3.model.GetObjectTorrentRequest;
import software.amazon.awssdk.services.s3.model.GetObjectTorrentResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.S3Object;

import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName AmazonService
 * @Author ZhanLei
 * @Date 2021/5/12 11:25
 */

@Service
public class AmazonAsyncService {

    private static final Logger logger = LoggerFactory.getLogger(AmazonAsyncService.class);

    @Autowired
    private S3AsyncClient s3AsyncClient;

    @Value("${aws.s3.bucket}")
    private String bucket;

    @Value("${aws.s3.path}")
    private String path;

    public void ops(Integer keys, String prefix) throws ExecutionException, InterruptedException {
        AtomicInteger num = new AtomicInteger();
        num.set(0);
        String token = "";
        List<S3Object> objs = new ArrayList<>();
        objs.add(S3Object.builder().build());
        // keys不能超过4500

        while (objs != null && !objs.isEmpty()) {
            ListObjectsV2Request request = ListObjectsV2Request.builder()
                    .bucket(bucket)
                    .maxKeys(keys)
                    .prefix(prefix)
                    .build();

            if (num.get() != 0) {
                request = ListObjectsV2Request.builder()
                        .bucket(bucket)
                        .continuationToken(token)
                        .maxKeys(keys)
                        .prefix(prefix)
                        .build();
            }

            CompletableFuture<ListObjectsV2Response> future =
                    s3AsyncClient.listObjectsV2(request);
            poolExecFor(future, num);

            ListObjectsV2Response listObjectsV2Response = future.get();
            objs = listObjectsV2Response.contents();
            token = listObjectsV2Response.nextContinuationToken();

        }

    }

    private void poolExecFor(CompletableFuture<ListObjectsV2Response> future, AtomicInteger num) {
        future.whenCompleteAsync((resp, err) -> {
            if (resp != null) {
                List<S3Object> contents = resp.contents();
                contents.parallelStream().forEach(s3Object -> {
                    int andIncrement = num.getAndIncrement();
//                    logger.info("Object keys. key: {}, number:{}", s3Object.key(), andIncrement);
                    execDownLoad(s3Object, andIncrement);
                });
            } else {
                err.printStackTrace();
            }

        });
    }

    private void execDownLoad(S3Object s3Object, Integer num) {
//        GetObjectRequest objectRequest = GetObjectRequest.builder()
//                .bucket(bucket)
//                .key(s3Object.key())
//                .build();

//        CompletableFuture<GetObjectResponse> futureGet = s3AsyncClient.getObject(objectRequest,
//                AsyncResponseTransformer.toFile(Paths.get(path + num)));


        GetObjectTorrentRequest objectRequest = GetObjectTorrentRequest.builder()
                .bucket(bucket)
                .key(s3Object.key())
                .build();

        CompletableFuture<GetObjectTorrentResponse> futureGet = s3AsyncClient.getObjectTorrent(objectRequest,
                AsyncResponseTransformer.toFile(Paths.get(path + num)));

        futureGet.whenCompleteAsync((resp, err) -> {

            if (resp != null) {
//                logger.info("Object downloaded. Details: {}", resp);
            } else {
                err.printStackTrace();
            }

        });

    }
}